/**
 * Copyright (c) 2014-2017, huli.com
 * All rights reserved.
 */
package com.xiaohujr.xuexue.rabbitmq;

import com.rabbitmq.client.*;
import com.xiaohujr.xuexue.rabbitmq.beans.QueueInfo;
import com.xiaohujr.xuexue.rabbitmq.constants.RabbitConstants;
import com.xiaohujr.xuexue.rabbitmq.interfaces.ConsumerHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 方法类.
 *
 * @author Xue Xue (xuexue1@huli.com)
 * @version $Id$
 * @since 2018/4/19
 */
public class Producer {

    /**
     * logger.
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);

    private static Connection conn = null;

    static {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitConstants.HOST);
        factory.setPort(RabbitConstants.PORT);
        factory.setUsername("root");
        factory.setPassword("root");
        try {
            conn = factory.newConnection();
        } catch (IOException | TimeoutException e) {
            LOGGER.error("connection create failed", e);
        }
    }

    public static Connection getConn() {
        return conn;
    }

    public static void sendMsgDirectly(QueueInfo queueInfo) {
        try (Channel channel = conn.createChannel()) {
            channel.exchangeDeclare(
                    queueInfo.getExchangeName(),
                    BuiltinExchangeType.DIRECT.getType(),
                    true,
                    false,
                    null
            );
            channel.queueDeclare(
                    queueInfo.getQueueName(),
                    true,
                    false,
                    false,
                    null
            );
            channel.queueBind(
                    queueInfo.getQueueName(),
                    queueInfo.getExchangeName(),
                    queueInfo.getRoutingKey()
            );
            channel.basicPublish(
                    RabbitConstants.DEFAULT_EXCHANGE,
                    RabbitConstants.DEFAULT_ROUTING_KEY,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,
                    queueInfo.getMessage().getBytes()
            );
        } catch (TimeoutException | IOException e) {
            LOGGER.error("channel of direct queue created failed!", e);
        }
    }

    public static void closeConn() {
        try {
            conn.close();
        } catch (IOException e) {
            LOGGER.error("close connection error!", e);
        }
    }

    public static void ConsumeMsg(String queueName,
                                  ConsumerHandler consumerHandler) {
        try (Channel channel = conn.createChannel()) {
            channel.basicQos(64);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)
                        throws IOException {
                    try {
                        consumerHandler.handleMsg(new String(body));
                    } catch (Exception e) {
                        LOGGER.error("handle msg error!", e);
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(queueName, consumer);
            TimeUnit.SECONDS.sleep(5);
        } catch (TimeoutException | IOException e) {
            LOGGER.error("channel of consume msg failed!", e);
        } catch (InterruptedException e) {
            LOGGER.info("sleep interrupted", e);
        }
    }
}
